#include "swoole_coroutine.h"
#include "swoole_coroutine_socket.h"
#include "swoole_coroutine_system.h"
#include "swoole_signal.h"

#include <iostream>
#include <list>
#include <algorithm>
#include <vector>

using swoole::Coroutine;
using swoole::coroutine::Socket;
using swoole::coroutine::System;
using namespace std;

list<string> q;
list<Socket *> slaves;
size_t qs;

int main(int argc, char **argv) {
    signal(SIGPIPE, SIG_IGN);

    swoole::coroutine::run([](void *arg) {
        Coroutine::create([](void *arg) {
            System::sleep(2.0);
            cout << "CO-1, sleep 2\n";
        });

        Coroutine::create([](void *arg) {
            System::sleep(1);
            cout << "CO-2, sleep 1\n";
        });

        Coroutine::create([](void *arg) {
            cout << "CO-3, listen tcp:0.0.0.0:9001\n";
            Socket s(SW_SOCK_TCP);
            s.bind("0.0.0.0", 9001);
            s.listen();

            while (1) {
                Socket *_client = s.accept();
                Coroutine::create(
                    [](void *arg) {
                        Socket *client = (Socket *) arg;
                        while (1) {
                            char buf[1024];
                            auto retval = client->recv(buf, sizeof(buf));
                            if (retval == 0) {
                                cout << "connection close\n";
                                break;
                            } else {
                                if (strncasecmp("push", buf, 4) == 0) {
                                    q.push_back(string(buf + 5, retval - 5));
                                    qs += retval - 5;
                                    string resp("OK\n");
                                    client->send(resp.c_str(), resp.length());

                                    for (auto it = slaves.begin(); it != slaves.end();) {
                                        auto sc = *it;
                                        auto n = sc->send(buf, retval);
                                        if (n <= 0) {
                                            it = slaves.erase(it);
                                            delete sc;
                                        } else {
                                            it++;
                                        }
                                    }
                                } else if (strncasecmp("pop", buf, 3) == 0) {
                                    if (q.empty()) {
                                        string resp("EMPTY\n");
                                        client->send(resp.c_str(), resp.length());
                                    } else {
                                        auto data = q.front();
                                        q.pop_front();
                                        qs -= data.length();
                                        client->send(data.c_str(), data.length());
                                    }
                                } else if (strncasecmp("stat", buf, 4) == 0) {
                                    char stat_buf[64];
                                    int n = snprintf(stat_buf, sizeof(stat_buf), "count=%ld,bytes=%ld\n", q.size(), qs);
                                    client->send(stat_buf, n);
                                } else {
                                    string resp("ERROR\n");
                                    client->send(resp.c_str(), resp.length());
                                }
                            }
                        }
                        delete client;
                    },
                    _client);
            }
        });

        Coroutine::create([](void *arg) {
            Socket s(SW_SOCK_TCP);
            s.bind("0.0.0.0", 9002);
            s.listen();
            while (1) {
                Socket *_client = s.accept();
                if (_client == nullptr) {
                    break;
                }
                for (auto data : q) {
                    _client->send(data.c_str(), data.length());
                }
                slaves.push_back(_client);
            }
        });
    });

    return 0;
}
